package com.chenwc.rocketmq.example;

import com.chenwc.rocketmq.prop.RocketMqConnectionProperties;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.List;

/**
 * DefaultLitePullConsumer是Apache RocketMQ提供的一种消费者类型，用于主动拉取消息进行消费
 * DefaultLitePullConsumer是RocketMQ 4.x版本及以后引入的一种PullConsumer类型，它提供了两种消息拉取方式：Subscribe（订阅）和Assign（分配）。
 * 与PushConsumer不同，PullConsumer需要用户自行控制消息的拉取和消费进度，提供了更灵活的消息消费机制。
 *
 * @author chenwc
 */
public class PullConsumerExample {
    private static final Logger logger = LoggerFactory.getLogger(PullConsumerExample.class);

    public static volatile boolean running = true;

    private PullConsumerExample() {
    }

    public static void main(String[] args) throws MQClientException {
        logger.info("Consumer start");
        RocketMqConnectionProperties properties = new RocketMqConnectionProperties();
        String topic = "TopicTest";
        // 配置 AccessKey 和 SecretKey
        SessionCredentials sessionCredentials = new SessionCredentials(properties.getAccessKey(), properties.getSecretKey());
        RPCHook aclHook = new AclClientRPCHook(sessionCredentials);
        // 创建一个Consumer
        DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("ConsumerGroup", aclHook);
        consumer.setNamesrvAddr(properties.getNamesrvAddr());

        // 从哪里开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 订阅Topic
        consumer.subscribe(topic, "*");
        // 设置PullTimeoutMillis， 拉取消息的时间间隔
        consumer.setPollTimeoutMillis(Duration.ofSeconds(3).toMillis());
        // 广播模式
        consumer.setMessageModel(MessageModel.BROADCASTING);
        // 启动消费对象
        consumer.start();

        while (running) {
            final List<MessageExt> messageExts = consumer.poll();
            logger.info("Received {} message(s)", messageExts.size());
            for (MessageExt message : messageExts) {
                final String messageId = message.getMsgId();
                byte[] body = message.getBody();
                logger.info("Received message, messageId={}, body={}", messageId, new String(body, StandardCharsets.UTF_8));
            }
        }
    }
}
